package org.example.jasper

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.functions._

object BZage {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("BZ")
    val sc = new SparkContext(conf)
    // 创建 SparkSession
    val spark = SparkSession.builder
      .appName("BZage")
      .getOrCreate()

    // 读取数据
    val inputPath = "src/main/java/org/datas/bz"
    val df = spark.read
      .option("header", "true")
      .csv(inputPath)

    // 将 "idade" 列转换为整数类型
    val dfWithAge = df.withColumn("idade", col("idade").cast("int"))

    // 过滤年龄在18到100岁之间的数据
    val filteredData = dfWithAge.filter(col("idade").between(18, 100))

    // 使用groupBy和agg统计每个年龄的车祸数量
    val result = filteredData
      .groupBy("idade")
      .agg(count("*").alias("accident_count"))
      .orderBy("idade")

    // 打印结果
    result.show()

    // 将结果导入到 MySQL 数据库中
    val mysqlUrl = "jdbc:mysql://localhost:3306/BZaccident"
    val mysqlProperties = new java.util.Properties()
    mysqlProperties.setProperty("user", "root")
    mysqlProperties.setProperty("password", "011216")
    mysqlProperties.setProperty("driver", "com.mysql.jdbc.Driver")

    result.write
      .mode(SaveMode.Overwrite) // 如果表已存在，覆盖之前的数据
      .jdbc(mysqlUrl, "BZage", mysqlProperties)
  }
}
